import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Objects;
import java.util.Properties;

@Slf4j
public class ControlService implements Runnable {

    public static final String CONTROL_TOPIC = "control";

    public static final KafkaConsumer<String, String> CONSUMER;


    static {
        Properties KAFKA_CONSUMER_PROPS = new Properties();
        KAFKA_CONSUMER_PROPS.put("bootstrap.servers", "47.92.23.67:9092");
        KAFKA_CONSUMER_PROPS.put("group.id", "general-group");
        KAFKA_CONSUMER_PROPS.put("max.poll.interval.ms", "3000000");
        KAFKA_CONSUMER_PROPS.put("auto.commit.interval.ms", "1000");
        KAFKA_CONSUMER_PROPS.put("auto.offset.reset", "latest");
        KAFKA_CONSUMER_PROPS.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KAFKA_CONSUMER_PROPS.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        CONSUMER = new KafkaConsumer<>(KAFKA_CONSUMER_PROPS);
        CONSUMER.subscribe(Collections.singletonList(CONTROL_TOPIC));
    }

    @Override
    public void run() {
        RaspberrySensor.FanThread fanThread = new RaspberrySensor.FanThread();
        while (!Thread.currentThread().isInterrupted()) { // 支持线程中断
            ConsumerRecords<String, String> records = CONSUMER.poll(Duration.ofMillis(10000));
            for (ConsumerRecord<String, String> record : records) {
                String consumerValue = record.value();
                log.info("offset = {}, key = {}, value = {}", record.offset(), record.key(), consumerValue);
                MessageBody messageBody = JacksonUtil.deserialize(consumerValue, MessageBody.class);
                if (Objects.isNull(messageBody)) {
                    continue;
                }
                if (messageBody.isGreenLed()) {
                    RaspberrySensor.ledGreenLightOn();
                } else {
                    RaspberrySensor.ledGreenLightOff();
                }
                if (messageBody.isOrangeLed()) {
                    RaspberrySensor.ledOrangeLightOn();
                } else {
                    RaspberrySensor.ledOrangeLightOff();
                }
                fanThread.refreshMessageBody(messageBody);
                TaskScheduler.scheduleGpioTask(fanThread);
                TaskScheduler.scheduleGpioTask(() -> RaspberrySensor.buzzer(messageBody.getBuzzer()));
                TaskScheduler.scheduleGpioTask(() -> RaspberrySensor.stepMotorRun(messageBody.getSteppingMotor()));
            }
        }
    }

    @Data
    public static class MessageBody {
        private boolean greenLed = false;
        private boolean orangeLed = false;
        private int fans = 0;
        private int buzzer = 0;
        private SteppingMotor steppingMotor;
    }

    @Data
    public static class SteppingMotor {
        private boolean clockwise = true;
        private double rpm = 0;
        private double circle = 0;
    }

    public static void main(String[] args) {
        new Thread(new ControlService()).start();
        SuperviseService.start();
    }
    /*
        #停止zookeeper
        bin/kafka-server-stop.sh
        #停止kafka
        bin/zookeeper-server-stop.sh
        #启动zookeeper
        bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
        #启动kafka
        bin/kafka-server-start.sh -daemon config/server.properties
        */

}
